package com.heima.kafka.sample;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/*
    流式处理
 */
public class KafkaStreamQuickStart {
    public static void main(String[] args) {
        //1.kafka配置信息
        Properties prop=new Properties();
        //1.1设置kafka的连接地址
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.135.128:9092");
        //1.2key和value的序列化
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"stream-quickStart");
        //2.创建stream构造器
        StreamsBuilder streamsBuilder=new StreamsBuilder();
        //流式计算
        StreamProcessor(streamsBuilder);
        //3.创建kafkaStream对象
        KafkaStreams kafkaStreams=new KafkaStreams(streamsBuilder.build(),prop);
        //4.开启流式计算
        kafkaStreams.start();
    }

    /**
     * 流式计算
     * 消息内容：hello kafka     hello zcfe
     * @param streamsBuilder
     */
    private static void StreamProcessor(StreamsBuilder streamsBuilder) {
        //创建KStream对象，同时指定从哪个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("zcfe-topic-input");
        /**
         * 处理消息的value值
         */
        stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value   ) {
                return Arrays.asList(value.split(" "));
            }

        })      //按照value进行聚合处理
                .groupBy((key,value)->value)
                //时间窗口 10s聚合一次数据
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //统计单词个数
                .count()
                //转换为stream对象
                .toStream()
                //这里处理完的消息key会变成某个消息的单词-所以需要用map进行处理
                .map((key,value)->{
                    System.out.println("key:"+key+",value:"+value);
                    return new KeyValue<>(key.key().toString(),value.toString());
                })
                //发送消息
                .to("zcfe-topic-out");

    }
}
